{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hosts.config is correctly read\n",
      "addresses: [['127.0.0.1', 10032], ['127.0.0.1', 10231], ['127.0.0.1', 10431], ['127.0.0.1', 10631]]\n",
      "['127.0.0.1', 10032]\n",
      "<class 'int'>\n",
      "['127.0.0.1', 10231]\n",
      "<class 'int'>\n",
      "['127.0.0.1', 10431]\n",
      "<class 'int'>\n",
      "['127.0.0.1', 10631]\n",
      "<class 'int'>\n",
      "sever: <Node at 0x7f0a28499290: _run>\n",
      "sever: <Node at 0x7f0a28499170: _run>\n",
      "sever: <Node at 0x7f0a28499290: _run>\n",
      "sever: <Node at 0x7f0a28499170: _run>\n"
     ]
    }
   ],
   "source": [
    "from gevent import  Greenlet\n",
    "import logging\n",
    "from gevent.queue import Queue\n",
    "\n",
    "class Node(Greenlet):\n",
    "\n",
    "    SEP = '\\r\\nSEP\\r\\n'\n",
    "\n",
    "    def __init__(self, port: int, ip: str, id: int, addresses_list: list, logger=None):\n",
    "        self.queue = Queue()\n",
    "        self.ip = ip\n",
    "        self.port = port\n",
    "        self.id = id\n",
    "        self.addresses_list = addresses_list\n",
    "        self.socks = [None for _ in self.addresses_list]\n",
    "        if logger is None:\n",
    "            self.logger = set_logger_of_node(self.id)\n",
    "        else:\n",
    "            self.logger = logger\n",
    "        Greenlet.__init__(self)\n",
    "\n",
    "    def _run(self):\n",
    "        self.logger.info(\"node %d starts to run...\" % self.id)\n",
    "        self._serve_forever()\n",
    "\n",
    "    def _handle_request(self, sock, address):\n",
    "\n",
    "        def _finish(e: Exception):\n",
    "            self.logger.error(\"node %d's server is closing...\" % self.id)\n",
    "            self.logger.error(str(e))\n",
    "            print(e)\n",
    "            print(\"node %d's server is closing...\" % self.id)\n",
    "            pass\n",
    "\n",
    "        buf = b''\n",
    "        try:\n",
    "            while True:\n",
    "                gevent.sleep(0)\n",
    "                buf += sock.recv(4096)\n",
    "                tmp = buf.split(self.SEP.encode('utf-8'), 1)\n",
    "                while len(tmp) == 2:\n",
    "                    buf = tmp[1]\n",
    "                    data = tmp[0]\n",
    "                    if data != '' and data:\n",
    "                        if data == 'ping'.encode('utf-8'):\n",
    "                            sock.sendall('pong'.encode('utf-8'))\n",
    "                            self.logger.info(\"node {} is pinging node {}...\".format(self._address_to_id(address), self.id))\n",
    "                        else:\n",
    "                            (j, o) = (self._address_to_id(address), pickle.loads(data))\n",
    "                            assert j in range(len(self.addresses_list))\n",
    "                            gevent.spawn(self.queue.put_nowait((j, o)))\n",
    "                    else:\n",
    "                        self.logger.error('syntax error messages')\n",
    "                        raise ValueError\n",
    "                    tmp = buf.split(self.SEP.encode('utf-8'), 1)\n",
    "        except Exception as e:\n",
    "            self.logger.error(str((e, traceback.print_exc())))\n",
    "            _finish(e)\n",
    "\n",
    "    def _serve_forever(self):\n",
    "        self.server_sock = socket.socket()\n",
    "        self.server_sock.bind((self.ip, self.port))\n",
    "        self.server_sock.listen(5)\n",
    "        while True:\n",
    "            sock, address = self.server_sock.accept()\n",
    "            gevent.spawn(self._handle_request, sock, address)\n",
    "            self.logger.info('node id %d accepts a new socket from node %d' % (self.id, self._address_to_id(address)))\n",
    "            gevent.sleep(0)\n",
    "\n",
    "    def _watchdog_deamon(self):\n",
    "        pass\n",
    "\n",
    "    def connect_all(self):\n",
    "        self.logger.info(\"node %d is fully meshing the network\" % self.id)\n",
    "        is_sock_connected = [False] * len(self.addresses_list)\n",
    "        while True:\n",
    "            try:\n",
    "                for j in range(len(self.addresses_list)):\n",
    "                    if not is_sock_connected[j]:\n",
    "                        is_sock_connected[j] = self._connect(j)\n",
    "                if all(is_sock_connected):\n",
    "                    break\n",
    "                time.sleep(1)\n",
    "            except Exception as e:\n",
    "                self.logger.info(str((e, traceback.print_exc())))\n",
    "\n",
    "    def _connect(self, j: int):\n",
    "        sock = socket.socket()\n",
    "        sock.bind((self.ip, self.port + j + 1))\n",
    "        try:\n",
    "            sock.connect(self.addresses_list[j])\n",
    "            sock.sendall(('ping' + self.SEP).encode('utf-8'))\n",
    "            pong = sock.recv(4096)\n",
    "        except Exception as e1:\n",
    "            return False\n",
    "            #print(e1)\n",
    "            #traceback.print_exc()\n",
    "        if pong.decode('utf-8') == 'pong':\n",
    "            self.logger.info(\"node {} is ponging node {}...\".format(j, self.id))\n",
    "            self.socks[j] = sock\n",
    "            return True\n",
    "        else:\n",
    "            self.logger.info(\"fails to build connect from {} to {}\".format(self.id, j))\n",
    "            return False\n",
    "\n",
    "    def _send(self, j: int, o: bytes):\n",
    "        msg = b''.join([o, self.SEP.encode('utf-8')])\n",
    "        try:\n",
    "            self.socks[j].sendall(msg)\n",
    "        except Exception as e1:\n",
    "            self.logger.error(\"fail to send msg\")\n",
    "            #print(\"fail to send msg\")\n",
    "            try:\n",
    "                self._connect(j)\n",
    "                self.socks[j].connect(self.addresses_list[j])\n",
    "                self.socks[j].sendall(msg)\n",
    "            except Exception as e2:\n",
    "                self.logger.error(str((e1, e2, traceback.print_exc())))\n",
    "\n",
    "    def send(self, j: int, o: object):\n",
    "        try:\n",
    "            self._send(j, pickle.dumps(o))\n",
    "        except Exception as e:\n",
    "            self.logger.error(str((\"problem objective when sending\", o)))\n",
    "            traceback.print_exc(e)\n",
    "\n",
    "    def _recv(self):\n",
    "        #time.sleep(0.001)\n",
    "        #try:\n",
    "        (i, o) = self.queue.get()\n",
    "        #print(\"node %d is receving: \" % self.id, (i, o))\n",
    "        return (i, o)\n",
    "        #except Exception as e:\n",
    "        #   print(e)\n",
    "        #   pass\n",
    "\n",
    "    def recv(self):\n",
    "        return self._recv()\n",
    "\n",
    "    def _address_to_id(self, address: tuple):\n",
    "        # print(address)\n",
    "        # print(self.addresses_list)\n",
    "        # assert address in self.addresses_list\n",
    "        for i in range(len(self.addresses_list)):\n",
    "            if address[0] != '127.0.0.1' and address[0] == self.addresses_list[i][0]:\n",
    "                return i\n",
    "        return int((address[1] - 10007) / 200)\n",
    "\n",
    "\n",
    "def set_logger_of_node(id: int):\n",
    "    logger = logging.getLogger(\"node-\"+str(id))\n",
    "    logger.setLevel(logging.DEBUG)\n",
    "    # logger.setLevel(logging.INFO)\n",
    "    formatter = logging.Formatter(\n",
    "        '%(asctime)s %(filename)s [line:%(lineno)d] %(funcName)s %(levelname)s %(message)s ')\n",
    "    if 'log' not in os.listdir(os.getcwd()):\n",
    "        os.mkdir(os.getcwd() + '/log')\n",
    "    # full_path = os.path.realpath(os.getcwd()) + '/log/' + \"node-\"+str(id) + \".log\"\n",
    "    full_path = os.path.realpath(os.getcwd())+'/log'+\"_test\"+str(id) + \".log\"\n",
    "    file_handler = logging.FileHandler(full_path)\n",
    "    file_handler.setFormatter(formatter)  # 可以通过setFormatter指定输出格式\n",
    "    logger.addHandler(file_handler)\n",
    "    return logger\n",
    "\n",
    "N = 4\n",
    "addresses = [None] * N\n",
    "try:\n",
    "    with open('hosts.config', 'r') as hosts:\n",
    "        for line in hosts:\n",
    "            params = line.split()\n",
    "            pid = eval(params[0])\n",
    "            ip = params[1]\n",
    "            port = eval(params[3])\n",
    "            # print(pid, ip, port)\n",
    "            if pid not in range(N):\n",
    "                continue\n",
    "            addresses[pid] = [ip, port]\n",
    "    # print(addresses)\n",
    "    assert all([node is not None for node in addresses])\n",
    "    print(\"hosts.config is correctly read\")\n",
    "    print('addresses:',addresses)\n",
    "except FileNotFoundError or AssertionError as e:\n",
    "    traceback.print_exc()\n",
    "addresses_list = addresses\n",
    "for i in range(N):\n",
    "    print(addresses_list[i])\n",
    "    print(type(addresses_list[i][1]))\n",
    "sever_1 = Node(id=1, ip=addresses_list[1][0], port=addresses_list[1][1], addresses_list=addresses_list, logger=set_logger_of_node(1))\n",
    "for i in range(N):\n",
    "    sever = Node(id=i, ip=addresses_list[i][0], port=addresses_list[i][1], addresses_list=addresses_list, logger=set_logger_of_node(i))\n",
    "    print('sever:',sever)\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.7.13 ('hbb': conda)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.13"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "7f5c99d094b399fa9e65cc4107643c6d5293cb8e1c21c8f7db515a0780c667bb"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
